package com.han.yrpc.core.runtime.nio;

import com.esotericsoftware.kryo.Kryo;
import com.esotericsoftware.kryo.io.Output;
import com.esotericsoftware.kryo.serializers.BeanSerializer;
import com.han.yrpc.core.registry.ServiceInfo;
import com.han.yrpc.core.runtime.YRequest;
import com.han.yrpc.core.runtime.YResponse;
import com.han.yrpc.core.runtime.YrpcClient;

import java.io.IOException;
import java.net.InetSocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.nio.channels.SocketChannel;
import java.util.Iterator;
import java.util.Set;

public class NioYrpcClient implements YrpcClient {

    private Selector selector;

    NioYrpcClient() {
        try {
            selector = Selector.open();

            while (true) {
                selector.select();
                Set<SelectionKey> keys = selector.selectedKeys();
                Iterator<SelectionKey> iterator = keys.iterator();
                while (iterator.hasNext()) {
                    SelectionKey key = iterator.next();
                    iterator.remove();

                    if (key.isReadable()) {
                        SocketChannel channel = (SocketChannel) key.channel();
                        ByteBuffer byteBuffer = ByteBuffer.allocate(1024);
                        int read = channel.read(byteBuffer);
                        if (read > 0) {
                            byteBuffer.flip();
                            byte[] bytes = new byte[byteBuffer.remaining()];
                            byteBuffer.get(bytes);
                            System.out.println(new String(bytes));
                        } else if (read < 0) {
                            key.cancel();
                            channel.close();
                        }
                    }
                }
            }
        } catch (IOException e) {
            e.printStackTrace();
        }
    }

    @Override
    public YResponse<Object> send(YRequest request) {
        ServiceInfo serviceInfo = request.getServiceInfo();

        try {
            SocketChannel socketChannel = SocketChannel.open();
            socketChannel.configureBlocking(false);

            InetSocketAddress socketAddress = new InetSocketAddress(serviceInfo.getHost(), serviceInfo.getPort());
            if (socketChannel.connect(socketAddress)) {
                socketChannel.register(selector, SelectionKey.OP_READ);
            } else {
                socketChannel.register(selector, SelectionKey.OP_CONNECT);
            }

            Kryo kryo = new Kryo();
            kryo.register(YRequest.class);
            Output output = new Output();
            kryo.writeObject(output, request);
            byte[] bytes = output.getBuffer();

            ByteBuffer writeBuffer = ByteBuffer.allocate(bytes.length);
            writeBuffer.put(bytes);
            writeBuffer.flip();
            socketChannel.write(writeBuffer);

        } catch (IOException e) {
            e.printStackTrace();
        }

        return null;
    }
}
